#include "MemCacheService.h"
#include <cassert>
#include "SvrBase.h"

int g_isFlushAllMemcache = 0;

sbase::CCriticalSection checkMutex_;


sbase::CCriticalSection mutexNewCount_;

__int64 g_newMemRequest = 0;
__int64 g_newMemCacheTask = 0;
__int64 g_delMemRequest  = 0;
__int64 g_delMemCacheTask = 0;


DWORD __stdcall MemCacheService::RunMemCacheTask(LPVOID lpThreadParameter)
{
	MemCacheService* service = static_cast<MemCacheService*>(lpThreadParameter);

	MemCacheClient mcClient;
	const std::vector<std::string>& mcServices = service->getServices();
	for (size_t i = 0; i < mcServices.size(); i++)
	{
		mcClient.AddServer(mcServices[i].c_str());
	}

	while (!service->quited())
	{

		{
			sbase::CSingleLock lock(&checkMutex_);
			if(g_isFlushAllMemcache)
			{
				int succCount = mcClient.FlushAll();

				printf("\nClean Memcache Succ [%d]\n",succCount);

				g_isFlushAllMemcache = 0;
			}
		}


		MemCacheTask* task = service->popTask();
		if (task == NULL)
		{
			Sleep(1);
			continue;
		}

		switch(task->RequestType)
		{
		case MC_ADD:
			mcClient.Add(task->Requests, task->RequestCount);
			break;

		case MC_SET:
			mcClient.Set(task->Requests, task->RequestCount);
			break;

		case MC_REPLACE:
			mcClient.Replace(task->Requests, task->RequestCount);
			break;

		case MC_APPEND:
			mcClient.Append(task->Requests, task->RequestCount);
			break;

		case MC_PREPEND:
			mcClient.Prepend(task->Requests, task->RequestCount);
			break;

		case MC_CAS:
			mcClient.CheckSet(task->Requests, task->RequestCount);
			break;

		case MC_GET:
			mcClient.Get(task->Requests, task->RequestCount);
			break;

		case MC_GETS:
			mcClient.Gets(task->Requests, task->RequestCount);
			break;

		case MC_DEL:
			mcClient.Del(task->Requests, task->RequestCount);
			break;
		}
		if (task->RequestCount > 0 && task->Requests[0].mResult == MCERR_NOSERVER)
		{
			fprintf(stderr, "Could not connect to MemCache Server.\n");
		}
		if (task->Handler != NULL)
		{
			service->pushFinishedTask(task);
		}
		else
		{
			extern sbase::CCriticalSection mutexNewCount_;
			sbase::CSingleLock tmplock(&mutexNewCount_);
			++g_delMemCacheTask;
			delete task;
		}
	}

	return 0;
}

MemCacheService::MemCacheService()
:quited_(false),bInit(false)
{

}

MemCacheService::~MemCacheService()
{
	quit();
	printf("MemCacheService::~MemCacheService: Waiting for worker thread to quit.\n");
	for (size_t i = 0; i < handlers_.size(); i++)
	{
		HANDLE h = handlers_[i];
		//::TerminateThread(h, 1L);
		//::WaitForSingleObject(h, INFINITE);
		CloseHandle(h);
	}
	printf("MemCacheService::~MemCacheService: Finished.\n");
	handlers_.clear();
}

bool MemCacheService::init(int threadNum, const std::vector<std::string>& services)
{
	services_ = services;
	if (threadNum <= 0)
	{
		threadNum = 1;
	}

	for (int i = 0; i < threadNum; i++)
	{
		DWORD id;
		HANDLE h = ::CreateThread (NULL, 0, MemCacheService::RunMemCacheTask,
			(void *)this, 0, &id);

		if (h == NULL)
		{
			fprintf(stderr, "MemCacheService::init: CreateThread Failed: %u.\n",
				GetLastError());
			return false;
		}
		handlers_.push_back(h);
	}

	bInit = true;
	return true;
}

bool MemCacheService::IsInit()
{
	return bInit;
}

void MemCacheService::quit()
{
	quited_ = true;
}


void MemCacheService::asyncQuery(MC_REQUEST_TYPE requestType, MemCacheClient::MemRequest* requests,
								 int requestCount, MemCacheHandler* handler)
{
	assert(requestType != NULL);

	MemCacheTask* task = new MemCacheTask;
	task->RequestType = requestType;
	task->Requests = requests;
	task->RequestCount = requestCount;
	task->Handler = handler;

	sbase::CSingleLock tmplock(&mutexNewCount_);
	++g_newMemRequest;
	++g_newMemCacheTask;

	sbase::CSingleLock lock(&mutex_);
	tasks_.push_back(task);
}

bool MemCacheService::tick()
{
	SBASE_TRY
	if (finishedTasks_.empty())
	{
		return false;
	}

	MemCacheTasks tasks;
	{
		sbase::CSingleLock lock(&mutex_);
		tasks.swap(finishedTasks_);
	}

	while(!tasks.empty())
	{
		MemCacheTask* task = tasks.front();
		tasks.pop_front();

		if (task->Handler != NULL)
		{
			task->Handler->HandleMemCache(task->Requests, task->RequestCount);
		}

		extern sbase::CCriticalSection mutexNewCount_;
		sbase::CSingleLock tmplock(&mutexNewCount_);

		++g_delMemCacheTask;

		delete task;
	}
	return true;
	SBASE_CATCH_DUMP("MemCacheService::tick()!")
	return false;
} 

MemCacheTask* MemCacheService::popTask()
{
	if (tasks_.empty())
	{
		return NULL;
	}

	sbase::CSingleLock lock(&mutex_);
	if (tasks_.empty())
	{
		return NULL;
	}

	MemCacheTask* task = tasks_.front();
	tasks_.pop_front();
	return task;
}

void MemCacheService::pushFinishedTask(MemCacheTask* task)
{
	sbase::CSingleLock lock(&mutex_);
	finishedTasks_.push_back(task);
}

